package com.open.source.kafka;

import java.nio.charset.StandardCharsets;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationContextAware;
import org.springframework.kafka.core.ConsumerFactory;

/**
 * @author ZonLen since on 2022/8/16 下午12:33
 */
public interface KafkaConsumerListener extends ApplicationRunner, ApplicationContextAware,
    DisposableBean {

  String kafkaTopic();

  default ConsumerFactory<Object, Object> consumerFactory() {
    return ApplicationContextHolder.getBean(ConsumerFactory.class);
  }

  default String deserializerEncoding() {
    return StandardCharsets.UTF_8.name();
  }

  default String getHeaderByKey(String key, ConsumerRecord<?, ?> record) {
    final Header header = record.headers().lastHeader(key);
    return new String(header.value());
  }
}
